package cn.ac.iie.di.ban.ycl.cleaner.receiver;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.util.concurrent.BlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-14 10:22:38
 * @version V1.0
 * @desc
 */
public class MessageSuccessReceiver extends REAbstractReceiveMessageHandler<REMessage> {

    final Logger LOG = LoggerFactory.getLogger(MessageSuccessReceiver.class);
    BlockingQueue<REMessage> queue;

    public MessageSuccessReceiver(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public boolean handle(REMessage message) {
        synchronized (MessageSuccessReceiver.class) {
            while (true) {
                boolean isSuccess = queue.offer(message);
                if (!isSuccess) {
                    LOG.warn("receiver queue is full,please increase handler thread "
                            + "or decrease mq consumer thread!");
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException ex) {
                    }
                } else {
                    LOG.info("push message num: {}",message.data.size());
                    break;
                }
            }
        }
        return true;
    }
}
